package com.ruicar.afs.cloud.common.mq.rabbit.init;

import cn.hutool.core.util.ReflectUtil;
import com.ruicar.afs.cloud.common.core.spring.DynamicBeanHelper;
import com.ruicar.afs.cloud.common.core.util.AfsResourcesScanner;
import com.ruicar.afs.cloud.common.core.util.SpringContextHolder;
import com.ruicar.afs.cloud.common.mq.rabbit.anno.receiver.AfsRabbitMqListener;
import com.ruicar.afs.cloud.common.mq.rabbit.anno.receiver.AfsRabbitMqReceiver;
import com.ruicar.afs.cloud.common.mq.rabbit.config.AfsRabbitMqProperties;
import com.ruicar.afs.cloud.common.mq.rabbit.config.rabbit.RabbitMqConfig;
import com.ruicar.afs.cloud.common.mq.rabbit.fallback.SendFallBack;
import com.ruicar.afs.cloud.common.mq.rabbit.helper.CacheInfo;
import com.ruicar.afs.cloud.common.mq.rabbit.helper.bean.ServerQueueRef;
import com.ruicar.afs.cloud.common.mq.rabbit.helper.bean.ServerTemplateRef;
import com.ruicar.afs.cloud.common.mq.rabbit.listener.ListenerFactory;
import com.ruicar.afs.cloud.common.mq.rabbit.message.AfsTransEntity;
import com.ruicar.afs.cloud.common.mq.rabbit.register.AfsRabbitMqSenderRegister;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;


/**
 * The type RabbitMqInitComponent
 * <p>Description: </p>
 *
 * @author Fzero
 * @version 1.0
 * @date Created on 2020.06.12 09:15:16
 * @since 1.0
 */
@Slf4j
public class RabbitMqInit {
    private ApplicationContext applicationContext;

    /**
     * Rabbit mq init component
     *
     * @param afsRabbitMqProperties afs rabbit mq properties
     */
    public RabbitMqInit(AfsRabbitMqProperties afsRabbitMqProperties, ApplicationContext applicationContext) {
        this.afsRabbitMqProperties = afsRabbitMqProperties;
        this.applicationContext = applicationContext;
        this.init();
    }

    /**
     * Afs rabbit mq properties
     * <p>Description: </p>
     */
    private AfsRabbitMqProperties afsRabbitMqProperties;

    /**
     * init *
     */
    public void init() {
        ListenerFactory.init(this.getClass().getClassLoader());
        DefaultListableBeanFactory autowireCapableBeanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
        Set<String> serverQueueNameSet = new HashSet<>();
        afsRabbitMqProperties
                .getServerList()
                .entrySet().stream().filter(entry-> entry.getValue().isEnable()).forEach(entry -> {
            DynamicBeanHelper.registerClass(entry.getKey(), CachingConnectionFactory.class, genConnectionFactoryConfig(entry.getValue()), autowireCapableBeanFactory, false, "");
            DynamicBeanHelper.registerClass(entry.getValue().getTemplateName(entry.getKey()), RabbitTemplate.class, getTemplateConfig(entry.getKey()), autowireCapableBeanFactory, false, "");
            DynamicBeanHelper.registerClass(entry.getValue().getListenerName(entry.getKey()), SimpleRabbitListenerContainerFactory.class, getRabbitListenerContainerFactoryConfig(entry.getKey(), entry.getValue()), autowireCapableBeanFactory, false, "");
            CacheInfo.CACHE_SERVER_TEMPLATE.put(entry.getKey(),
                    ServerTemplateRef.builder()
                            .serverKey(entry.getKey())
                            .springListenerName(entry.getValue().getListenerName(entry.getKey()))
                            .templateName(entry.getValue().getTemplateName(entry.getKey()))
                            .rabbitTemplate((RabbitTemplate) applicationContext.getBean(entry.getValue().getTemplateName(entry.getKey()))).build());
            log.info("rabbit MQ 初始化成功\n配置信息:{}\n服务器别名:{}\ntemplate名称:{}\n监听器名称:{}", entry.getValue(), entry.getKey(), entry.getValue().getTemplateName(entry.getKey()), entry.getValue().getListenerName(entry.getKey()));
        });
        AfsResourcesScanner
                .scanClassByAnnotation(new String[]{"com.ruicar.afs.cloud"}, AfsRabbitMqListener.class)
                .forEach(clazz -> {
                    AfsRabbitMqListener afsRabbitMqListener = AnnotationUtils.findAnnotation(clazz, AfsRabbitMqListener.class);
                    boolean init = true;
                    if(!StringUtils.isEmpty(afsRabbitMqListener.enableOn())&&!applicationContext.getEnvironment().containsProperty(afsRabbitMqListener.enableOn().replace("${", "").replace("}", ""))){
                        init = false;
                    }
                    if (afsRabbitMqListener.afsListener()&&init) {
                        if(init) {
                            String serverAlias = afsRabbitMqListener.serverAlias();
                            String queueName = afsRabbitMqListener.queueName();
                            if (serverAlias.startsWith("${")) {
                                Assert.isTrue(applicationContext.getEnvironment().containsProperty(serverAlias.replace("${", "").replace("}", "")), "参数" + serverAlias + "不存在");
                                serverAlias = applicationContext.getEnvironment().getProperty(serverAlias.replace("${", "").replace("}", ""));
                            }
                            if (queueName.startsWith("${")) {
                                Assert.isTrue(applicationContext.getEnvironment().containsProperty(queueName.replace("${", "").replace("}", "")), "参数" + queueName + "不存在");
                                queueName = applicationContext.getEnvironment().getProperty(queueName.replace("${", "").replace("}", ""));
                            }
                            Assert.isTrue(!serverQueueNameSet.contains(serverAlias + "-" + queueName), "重复定义[" + serverAlias + "-" + queueName + "]监听队列");
                            Class listenerClass = ListenerFactory.makeListener(serverAlias, queueName, CacheInfo.CACHE_SERVER_TEMPLATE.get(serverAlias).getSpringListenerName());
                            DynamicBeanHelper.registerClass(listenerClass.getSimpleName(), listenerClass, null, autowireCapableBeanFactory, false, "");
                            applicationContext.getBean(listenerClass.getSimpleName());
                            log.info("\n rabbit MQ afs平台监听器初始化成功\n服务器别名:{}\n队列名称:{}", serverAlias, queueName);
                        }
                    } else if(init) {
                        String beanName = StringUtils.isEmpty(afsRabbitMqListener.beanName()) ? clazz.getSimpleName() : afsRabbitMqListener.beanName();
                        DynamicBeanHelper.registerClass(beanName, clazz, null, autowireCapableBeanFactory, false, "");
                        Method[] methods = ReflectUtil.getMethods(clazz);
                        for (Method method : methods) {
                            AfsRabbitMqReceiver afsRabbitMqReceiver = AnnotationUtils.findAnnotation(method, AfsRabbitMqReceiver.class);
                            if (afsRabbitMqReceiver != null) {
                                boolean afsRabbitMqReceiverInit = true;
                                if(!StringUtils.isEmpty(afsRabbitMqReceiver.enableOn())&&!applicationContext.getEnvironment().containsProperty(afsRabbitMqReceiver.enableOn().replace("${", "").replace("}", ""))){
                                    afsRabbitMqReceiverInit = false;
                                }
                                if(afsRabbitMqReceiverInit) {
                                    String serverAlias = afsRabbitMqReceiver.serverAlias();
                                    String queueName = afsRabbitMqReceiver.queueName();
                                    if (serverAlias.startsWith("${")) {
                                        Assert.isTrue(applicationContext.getEnvironment().containsProperty(serverAlias.replace("${", "").replace("}", "")), "参数" + serverAlias + "不存在");
                                        serverAlias = applicationContext.getEnvironment().getProperty(serverAlias.replace("${", "").replace("}", ""));
                                    }
                                    if (queueName.startsWith("${")) {
                                        Assert.isTrue(applicationContext.getEnvironment().containsProperty(queueName.replace("${", "").replace("}", "")), "参数" + queueName + "不存在");
                                        queueName = applicationContext.getEnvironment().getProperty(queueName.replace("${", "").replace("}", ""));
                                    }
                                    Assert.isTrue(!serverQueueNameSet.contains(serverAlias + "-" + queueName), "重复定义[" + serverAlias + "-" + queueName + "]监听队列");
                                    serverQueueNameSet.add(serverAlias + "-" + queueName);
                                    Assert.isTrue(method.getParameterCount() == 1, "监听MQ方法必须为一个参数");
                                    ServerQueueRef serverQueueRef = ServerQueueRef.builder()
                                            .method(method)
                                            .object(applicationContext.getBean(beanName))
                                            .requeue(afsRabbitMqReceiver.requeue())
                                            .targetClass(afsRabbitMqReceiver.targetClass())
                                            .queueName(queueName)
                                            .serverAlias(serverAlias)
                                            .build();
                                    if (method.getParameterTypes()[0].equals(AfsTransEntity.class)) {
                                        serverQueueRef.setTransEntity(Boolean.TRUE);
                                        Assert.isTrue(!(((ParameterizedTypeImpl) method.getGenericParameterTypes()[0]).getActualTypeArguments() == null && afsRabbitMqReceiver.targetClass().equals(Void.class)), "监听器反序列化类型未定义,服务别名[" + serverAlias + "],队列名[" + queueName + "]");
                                        Assert.isTrue(!(((ParameterizedTypeImpl) method.getGenericParameterTypes()[0]).getActualTypeArguments().length == 0 && afsRabbitMqReceiver.targetClass().equals(Void.class)), "监听器反序列化类型未定义,服务别名[" + serverAlias + "],队列名[" + queueName + "]");
                                        if (((ParameterizedTypeImpl) method.getGenericParameterTypes()[0]).getActualTypeArguments().length == 0) {
                                            serverQueueRef.setTransBodyType(afsRabbitMqReceiver.targetClass());
                                        } else {
                                            try {
                                                Class transBodyType = Class.forName((((ParameterizedTypeImpl) method.getGenericParameterTypes()[0]).getActualTypeArguments()[0]).getTypeName());
                                                Assert.isTrue(!(((ParameterizedTypeImpl) method.getGenericParameterTypes()[0]).getActualTypeArguments() == null && afsRabbitMqReceiver.targetClass().equals(Void.class)), "监听器反序列化类型未定义,服务别名[" + serverAlias + "],队列名[" + queueName + "]");
                                                serverQueueRef.setTransBodyType(transBodyType);
                                            } catch (ClassNotFoundException e) {
                                                log.error("", e);
                                                throw new BeanDefinitionValidationException("监听器反序列化泛型类出错,服务别名[" + serverAlias + "],队列名[" + queueName + "]");
                                            }
                                        }
                                    } else {
                                        serverQueueRef.setTransEntity(Boolean.FALSE);
                                    }
                                    CacheInfo.CACHE_SERVER_QUEUE.put(serverAlias + "-" + queueName, serverQueueRef);
                                    Class listenerClass = ListenerFactory.makeListener(serverAlias, queueName, CacheInfo.CACHE_SERVER_TEMPLATE.get(serverAlias).getSpringListenerName());
                                    DynamicBeanHelper.registerClass(listenerClass.getSimpleName(), listenerClass, null, autowireCapableBeanFactory, false, "");
                                    applicationContext.getBean(listenerClass.getSimpleName());
                                    log.info("rabbit MQ方法级别:监听器初始化成功\n服务器别名:{}\n队列名称:{}", serverAlias, queueName);
                                }
                            }
                        }
                    }
                });


        AfsRabbitMqSenderRegister.fallBackDefineMap.entrySet().forEach(entry -> {
            try {
                String name = entry.getKey();
                log.info("mq回调-{}-{}",entry.getKey(),entry.getValue());
                if (name.startsWith("${")) {
                    name = applicationContext.getEnvironment().getProperty(name.replace("${", "").replace("}", ""));
                }
                DynamicBeanHelper.registerClass(name, entry.getValue(), null, autowireCapableBeanFactory, false, "");
            } catch (Exception e) {
                log.error("初始化mq调用端回调类失败", e);
            }
        });
    }


    /**
     * Gen connection factory config list
     *
     * @param rabbitMqConfig rabbit mq config
     * @return the list
     */
    private List<DynamicBeanHelper.BeanParamConfig> genConnectionFactoryConfig(RabbitMqConfig rabbitMqConfig) {
        List<DynamicBeanHelper.BeanParamConfig> paramConfigList = new ArrayList<>();
        if (!StringUtils.isEmpty(rabbitMqConfig.getClusterAddressesList())) {
            paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("addresses").value(rabbitMqConfig.getClusterAddressesList()).build());
        } else {
            paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("host").value(rabbitMqConfig.getHost()).build());
            paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("port").value(rabbitMqConfig.getPort()).build());
        }
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("username").value(rabbitMqConfig.getUsername()).build());
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("password").value(rabbitMqConfig.getPassword()).build());
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("virtualHost").value(rabbitMqConfig.getVirtualHost()).build());
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("publisherConfirms").value(true).build());
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("publisherReturns").value(true).build());
        return paramConfigList;

    }

    /**
     * Get template config list
     *
     * @param key key
     * @return the list
     */
    private List<DynamicBeanHelper.BeanParamConfig> getTemplateConfig(final String key) {
        List<DynamicBeanHelper.BeanParamConfig> paramConfigList = new ArrayList<>();
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.CON_REF).refBeanName(key).build());
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("mandatory").value(true).build());
        RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            private final Logger LOGGER = LoggerFactory.getLogger("mq-ack-callback-" + key);

            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (!ack) {
                    LOGGER.info("{} 发送RabbitMQ消息 ack确认 失败: [{}]", key, cause);
                } else {
                    LOGGER.info("{} 发送RabbitMQ消息 ack确认 成功", key);
                }
            }
        };
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("confirmCallback").value(confirmCallback).build());
        RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
            private final Logger LOGGER = LoggerFactory.getLogger("mq-return-callback-" + key);

            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                if (replyCode != 200) {
                    SpringContextHolder.getApplicationContext().getBean(routingKey, SendFallBack.class).errorHandler(replyCode, exchange, routingKey, new String(message.getBody()));
                    LOGGER.error("{} 发送RabbitMQ消息returnedMessage，出现异常，Exchange不存在或发送至Exchange却没有发送到Queue中，message：[{}], code[{}], s[{}], exchange[{}], routingKey[{}]", key, new String(message.getBody()), replyCode, replyText, exchange, routingKey);
                } else {
                    SpringContextHolder.getApplicationContext().getBean(routingKey, SendFallBack.class).successHandler(replyCode, exchange, routingKey, new String(message.getBody()));
                }
            }
        };
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("returnCallback").value(returnCallback).build());
        return paramConfigList;
    }

    /**
     * Get rabbit listener container factory config list
     *
     * @param key            key
     * @param rabbitMqConfig rabbit mq config
     * @return the list
     */
    private List<DynamicBeanHelper.BeanParamConfig> getRabbitListenerContainerFactoryConfig(String key, RabbitMqConfig rabbitMqConfig) {
        List<DynamicBeanHelper.BeanParamConfig> paramConfigList = new ArrayList<>();
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_REF).propertyName("connectionFactory").refBeanName(key).build());
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("prefetchCount").value(rabbitMqConfig.getPrefetch()).build());
        paramConfigList.add(DynamicBeanHelper.BeanParamConfig.builder().typeEnum(DynamicBeanHelper.ParamTypeEnum.PRO_OBJECT).propertyName("acknowledgeMode").value(AcknowledgeMode.MANUAL).build());
        return paramConfigList;
    }
}
